package io.kimmking.kmq.demo;


import io.kimmking.kmq.core.KmqBroker;
import io.kimmking.kmq.core.KmqConsumer;
import io.kimmking.kmq.core.KmqMessage;
import io.kimmking.kmq.core.KmqProducer;
import lombok.SneakyThrows;

import java.io.IOException;

public class KmqDemo {

    @SneakyThrows
    public static void main(String[] args) {
        String topic = "kk.test";
        KmqBroker broker = new KmqBroker();
        broker.createTopic(topic);
        KmqConsumer consumer = broker.createConsumer();
        consumer.subscribe(topic);
        KmqProducer producer = broker.createProducer();

        final boolean[] flag = new boolean[1];
        flag[0] = true;

        Thread consumerThread = new Thread(() -> {
            while (flag[0]) {
                // try {
                //     // 不加这个消费线程假死
                //     Thread.sleep(0);
                // } catch (InterruptedException e) {
                //     e.printStackTrace();
                // }
                KmqMessage message = consumer.poll(100);
                if(null != message) {
                    System.out.println("message：" + message.getBody());
                }
            }
            System.out.println("程序退出。");
        });
        consumerThread.start();

       System.out.println("点击任何键，发送一条消息；点击q或e，退出程序。");
       while (true) {
           char c = 0;
           try {
               c = (char) System.in.read();
           } catch (IOException e) {
               e.printStackTrace();
           }
           if( c == 'q' || c == 'e'){
               flag[0] = false;
               break;
           }
           if(c > 20) {
               Order order = new Order(1020000L + c, System.currentTimeMillis(), "USD2CNY", 6.52d);
               boolean tag = producer.send(topic, new KmqMessage(topic,null, order));
               System.out.println("producer result:" + tag);
           }
       }

    }
}
